package fun.pengzh.demo.kafka.sample4;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.DltHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.retry.annotation.Backoff;

/**
 * 基于topic的消息重试机制
 */
@SpringBootApplication
public class KafkaDemoApplication {

    private final static Logger logger = LoggerFactory.getLogger(KafkaDemoApplication.class);

    public static void main(String[] args) throws Exception {
        SpringApplication.run(KafkaDemoApplication.class, args);
    }


    @RetryableTopic(attempts = "5", backoff = @Backoff(delay = 2_000, maxDelay = 10_000, multiplier = 2))
    @KafkaListener(groupId = "fooGroup4", topics = "topic4")
    public void listener4(String msg, @Header(KafkaHeaders.REPLY_TOPIC) String topic, @Header(KafkaHeaders.OFFSET) long offset) {
        logger.info("listener4 Received: topic={}, offset={}, msg={}", topic, offset, msg);
        if (msg.startsWith("fail")) {
            throw new RuntimeException("listener4 consume failed ");
        }
    }

    @DltHandler
    public void listenDlt(String msg, @Header(KafkaHeaders.REPLY_TOPIC) String topic, @Header(KafkaHeaders.OFFSET) long offset) {
        logger.info("listener4 Received: topic={}, offset={}, msg={}", topic, offset, msg);
    }

}

